package com.hmall.delay;

import com.alibaba.fastjson.JSONObject;
import com.hdmall.business.common.common.ServerResponse;
import com.hmall.delay.model.Route;
import com.hmall.controller.JpushController;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import javax.annotation.Resource;
import java.io.IOException;
import java.text.ParseException;
import java.util.Date;
import java.util.List;
import java.util.Set;

@Component
public class MsgTask {

    @Resource
    private RedisMQ redisMQ;
    // @Value("${mq.list.first}") private String MQ_LIST_FIRST;

    private List<Route> routes;

    @Autowired
    JpushController jpushController;
    /**
     * 默认监听数量，对应监听zset队列前多少个元素
     */
    private static final int DEFAUT_MONITOR = 10;

    /**
     * 每次监听queue中元素的数量，可配置
     */
    private int monitorCount = DEFAUT_MONITOR;

    @Autowired
    JedisPool jedisPool;


    //消费在list里的消息
    @Scheduled(cron="*/30 * * * * *")
    public void sendMsg() throws IOException, ParseException {
        // 消费
//        Calendar calendar = Calendar.getInstance();
//        calendar.setTime(new Date());
//        String date = calendar.getTime().toString();
//        System.out.println("当前时间为 " + date);
        List<String> msgs = redisMQ.consume(redisMQ.getRoutes().get(0).getList());
        int len;
        if (null != msgs && 0 < (len = msgs.size())) {
            // 将每一条消息转为JSONObject
            JSONObject jObj;
            for (int i = 0; i < len; i++) {
                if (!StringUtils.isEmpty(msgs.get(i))) {
                    jObj = JSONObject.parseObject(msgs.get(i));
                    String message = (String) jObj.get("msg");
                    Integer customerUserId = (Integer) jObj.get("customerUserId");
                    Date date = (Date) jObj.get("date");
                    String value = (String) jObj.get("value");
                    Integer status = (Integer) jObj.get("status");
                    // 取出消息
                    //————————————————
                    //消息推送
                    ServerResponse response = jpushController.backPushMessage(customerUserId, status, value);
                    System.out.println(response);
                }
            }
        }
    }


    /**
     * 消息队列监听器<br>
     * 监听所有路由器，将消息队列中的消息路由到待消费列表
     */
    //将zset有序集合放到list里面
    @Scheduled(cron="*/20 * * * * *")
    public void monitor() {
        routes = redisMQ.getRoutes();
        Jedis jedis = jedisPool.getResource();
        // 获取消息路由
        int route_size;
        if (null == routes || 1 > (route_size = routes.size())) {
            return;
        }
        String queue, list;
        Set<String> set;
        for (int i = 0; i < route_size; i++) {
            queue = routes.get(i).getQueue();
            list = routes.get(i).getList();
            set = jedis.zrevrange(queue, 0, monitorCount);
            if (null != set) {
                long current = System.currentTimeMillis();
                long score;
                for (String id : set) {
                    score = jedis.zscore(queue, id).longValue();
                    if (current >= score) {
                        // 添加到list
                        jedis.rpush(list, id);

                        // 删除queue中的元素
                        redisMQ.deMessage(queue, id);
                    }
                }
            }
        }
        jedis.close();
    }

    public List<Route> getRoutes() {
        return routes;
    }

    public void setRoutes(List<Route> routes) {
        this.routes = routes;
    }
}